package com.aispeech.task.mgr.mapper;

import com.aispeech.task.lasr.event.EventType;
import com.aispeech.task.mgr.event.TaskEvent;
import com.aispeech.task.mgr.fsm.State;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

@Slf4j
public class StateMachineMapper implements MapFunction<TaskEvent, TaskEvent> {
	private State defaultState;

	public StateMachineMapper(State initialState) {
		defaultState = initialState;
	}

	@Override
	public TaskEvent map(TaskEvent event) {
		State state = defaultState;

		while (true) {
			Tuple2<State, TaskEvent> next = state.transition(event);

			log.warn("[ map ] event: {}, state: {}, next state: {}", event, state, next.f0);

			if (next.f0 == State.Invalid) {
				event.setType(EventType.Error);
				return event;
			} else if (next.f0.isTerminal()) {
				return next.f1;
			} else {
				event = next.f1;
				state = next.f0;
			}
		}
	}
}
